package beautiful.butterfly.drds.binlog.handlers;

import beautiful.butterfly.drds.binlog.connection.ConnectionAttributes;
import beautiful.butterfly.drds.binlog.consumer.ConsumerMananger;
import beautiful.butterfly.drds.binlog.event.EventData;
import beautiful.butterfly.drds.binlog.event.EventHeader;
import beautiful.butterfly.drds.binlog.event.EventType;
import beautiful.butterfly.drds.binlog.event.RowEventData;
import beautiful.butterfly.drds.binlog.event.data.RotateEventData;
import beautiful.butterfly.drds.binlog.netty.server.NioSocketChannelExtend;
import beautiful.butterfly.drds.binlog.parser.*;
import beautiful.butterfly.drds.binlog.snapshot.SnapShot;
import beautiful.butterfly.drds.binlog.snapshot.SnapShotType;
import beautiful.butterfly.drds.binlog.type.ParallelType;
import beautiful.butterfly.drds.binlog.utils.ByteUtils;
import beautiful.butterfly.drds.binlog.utils.Constants;
import beautiful.butterfly.drds.binlog.utils.Nettys;
import beautiful.butterfly.drds.binlog.utils.TimeUtils;
import beautiful.butterfly.drds.binlog.zookeeper.ZooKeepers;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

@Slf4j
public class BinlogEventParseChannelInboundHandler extends SimpleChannelInboundHandler<ByteBuf>
{
    // http://dev.mysql.com/doc/internals/en/binlog-event.html

    private static final EventType[] EVENT_TYPES = EventType.values();
    private boolean enterLargeBlockPacket = false;// 暂时不支持这种情况,TODO
    private HashMap<EventType, EventDataParser> eventTypeToEventDataParserMap = new HashMap<EventType, EventDataParser>();
    // private long totalIOEvent = 0;

    public BinlogEventParseChannelInboundHandler()
    {
        // null_event
        eventTypeToEventDataParserMap.put(EventType.unknown, new NullEventDataParser());
        // rotate
        eventTypeToEventDataParserMap.put(EventType.rotate, new RotateEventDataParser());
        // format_description
        eventTypeToEventDataParserMap.put(EventType.format_description, new FormatDescriptionParser());
        // query
        eventTypeToEventDataParserMap.put(EventType.query, new QueryEventDataParser());
        // table_map
        eventTypeToEventDataParserMap.put(EventType.table_map, new TableMapEventDataParser());
        // delete_rows
        eventTypeToEventDataParserMap.put(EventType.delete_rows, new DeleteRowsEventDataParser(false));
        // ext_delete_rows
        eventTypeToEventDataParserMap.put(EventType.ext_delete_rows, new DeleteRowsEventDataParser(true));
        // write_rows
        eventTypeToEventDataParserMap.put(EventType.write_rows, new WriteRowsEventDataParser(false));
        // ext_write_rows
        eventTypeToEventDataParserMap.put(EventType.ext_write_rows, new WriteRowsEventDataParser(true));
        // update_rows
        eventTypeToEventDataParserMap.put(EventType.update_rows, new UpdateRowsEventDataParser(false));
        // ext_update_rows
        eventTypeToEventDataParserMap.put(EventType.ext_update_rows, new UpdateRowsEventDataParser(true));
        // xid
        eventTypeToEventDataParserMap.put(EventType.xid, new XidEventDataParser());
        // rows_query
        eventTypeToEventDataParserMap.put(EventType.rows_query, new RowsQueryEventDataParser());
    }

    private void handleRotate(EventType eventType,//
                              EventHeader eventHeader,//
                              EventData eventData,//
                              ConnectionAttributes connectionAttributes, //
                              ChannelHandlerContext channelHandlerContext) throws Exception//
    {
        // newValue
        RotateEventData rotateEventData = (RotateEventData) eventData;
        String newBinlogFileName = rotateEventData.getBinlogFilename();
        long newBinlogPosition = rotateEventData.getBinlogFilePosition();
        String newValue = newBinlogFileName + ":" + newBinlogPosition;
        // localValue
        String localValue = connectionAttributes.getBinlogFileName() + ":" + connectionAttributes.getBinlogFilePosition();
        // 条件更新
        if (false == newValue.equals(localValue))
        {
            // 必须及时修改，别忘记了外面还有个historyZK路径需要修改的，所以要及时更新
            connectionAttributes.updateBinlogFileNameAndPosition(newBinlogFileName, newBinlogPosition);
            // 修改为放入taskThread线程，而不是立即更新到ZK,因为更新ZK很耗费时间
            // 挂载到队尾
            // 因为是AtomicReference，所以保证了线程安全性
            SnapShot snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.rotate, false, null, null,
                    null, newValue + ":" + System.currentTimeMillis(), null, null);
            connectionAttributes.addSnapShotToTail(snapShot);
            log.info("receive rotate event,not same as local,[BinlogEventParseChannelInboundHandler] send RequestBinaryLogCommand: "
                    + newBinlogFileName + ":" + newBinlogPosition);
            // 测试发现也可以不发送此命令，就取消了，减少线程互斥带来的问题
            // new RequestBinaryLogCommand(myAttributes.getClientId(),
            // newBinlogFileName, newBinlogPosition)
            // .write(context);
        } else
        {
            // 直接忽略,主要是启动时有一个默认的rotate事件
            log.info("recive rotate event,the same as local data,omit");
        }

    }

    private void handleIO(EventType eventType, EventHeader eventHeader, EventData eventData,
                          ConnectionAttributes connectionAttributes, ChannelHandlerContext channelHandlerContext) throws Exception
    {
        log.info(eventType.toString());
        // 1)存储成功了就更新内存变量
        connectionAttributes.updateBinlogFileNameAndPosition(connectionAttributes.getBinlogFileName(), eventHeader.getNextPosition());
        // 2)存储快照
        if (eventHeader.getNextPosition() > 0)
        {
            RowEventData rowEventData = (RowEventData) eventData;
            String databaseName = rowEventData.getDatabaseName();
            String tableName = rowEventData.getTableName();
            boolean accepted = connectionAttributes.acceptByFilter(databaseName, tableName);
            ArrayList<JSONObject> datas = accepted ? eventData.toJson() : new ArrayList<JSONObject>();
            String zkValues = connectionAttributes.getBinlogFileName() + ":" + connectionAttributes.getBinlogFilePosition() + ":"
                    + System.currentTimeMillis();
            SnapShot snapShot;
            ArrayList<JSONObject> tmpList;
            // 准备工作完毕
            if (false == accepted)
            {// 需要丢弃的，accepted保证不处理,consumer是不会处理的
                snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.io, accepted, databaseName, tableName,
                        datas, zkValues, ParallelType.table, null);
                ConsumerMananger.getInstance().addSnapShot(snapShot, connectionAttributes);
                connectionAttributes.addSnapShotToTail(snapShot);
            } else
            {// 确实需要处理的
                // 根据是否有主键来决定按表加速还是按行加速
                ArrayList<String> primaryKeyColumnNameList = connectionAttributes.getPrimaryKeyColumnNameList(databaseName, tableName);

                // 是否接受
                if (null == primaryKeyColumnNameList || primaryKeyColumnNameList.isEmpty())
                {// 按表复制-退化
                    snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.io, accepted, databaseName, tableName,
                            datas, zkValues, ParallelType.table, null);
                    ConsumerMananger.getInstance().addSnapShot(snapShot, connectionAttributes);
                    connectionAttributes.addSnapShotToTail(snapShot);
                } else
                {// 按行复制-加速-分拆，最后一条带上ZK信息，其它不带
                    int index = 0;
                    int size = datas.size();
                    for (JSONObject data : datas)
                    {
                        tmpList = new ArrayList<JSONObject>();
                        tmpList.add(data);
                        index++;
                        if (index != size)
                        {
                            snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.io, accepted, databaseName,
                                    tableName, tmpList, null, ParallelType.row, primaryKeyColumnNameList);
                        } else
                        {// 最后一条，带上zk
                            snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.io, accepted, databaseName,
                                    tableName, tmpList, zkValues, ParallelType.row, primaryKeyColumnNameList);
                        }
                        ConsumerMananger.getInstance().addSnapShot(snapShot, connectionAttributes);
                        connectionAttributes.addSnapShotToTail(snapShot);
                    }
                }
            }
            // LoggerUtils.debug(logger, "insert into task thread queue
            // succeed");
        } else
        {
            log.info("nextBinlogPosition:" + eventHeader.getNextPosition());
            throw new Exception("io data,but nextBinlogPosition is 0");
        }

    }

    private void handlerDefault(EventType eventType, EventHeader eventHeader, EventData eventData,
                                ConnectionAttributes connectionAttributes, ChannelHandlerContext channelHandlerContext)
    {
        // 其它不处理的事情都在这里
        // 这里其实不能写业务[比如更新本地内存位置]，否则下次获取会出错
    }

    private void handleXID(//
                           EventType eventType, //
                           EventHeader eventHeader,//
                           EventData eventData,//
                           ConnectionAttributes connectionAttributes,//
                           ChannelHandlerContext channelHandlerContext)//
    {
        // 直接加到尾巴上，作为一个完整事件的结束
        SnapShot snapShot = new SnapShot(connectionAttributes.getGlobalValid(), SnapShotType.xid, false, null, null, null, null,
                null, null);
        connectionAttributes.addSnapShotToTail(snapShot);
    }

    private void handle(EventType eventType, EventHeader eventHeader, EventData eventData, ConnectionAttributes connectionAttributes,
                        ChannelHandlerContext channelHandlerContext) throws Exception
    {
        log.info("enter handle");
        // 保证参数OK
        if (null == eventType || null == eventData || null == connectionAttributes)
        {
            return;
        }
        switch (eventType)
        {
            case rotate:
                handleRotate(eventType, eventHeader, eventData, connectionAttributes, channelHandlerContext);
                break;
            case write_rows:
            case ext_write_rows:
            case update_rows:
            case ext_update_rows:
            case delete_rows:
            case ext_delete_rows:
                handleIO(eventType, eventHeader, eventData, connectionAttributes, channelHandlerContext);
                break;
            case xid:
                handleXID(eventType, eventHeader, eventData, connectionAttributes, channelHandlerContext);
                break;
            default:// 其它事件如何处理
                handlerDefault(eventType, eventHeader, eventData, connectionAttributes, channelHandlerContext);
                break;
        }
        log.info("leave handle");
    }

    private void handleZKHistoryRecord(EventHeader eventHeader, ConnectionAttributes connectionAttributes)
    {
        // 每一条都会处理,不区分具体事件类型
        // LoggerUtils.debug(logger, "enter handleZKHistoryRecord");
        long timestamp = eventHeader.getTimestamp();
        log.info("timestamp:" + timestamp);
        if (eventHeader.getEventType() == EventType.rotate && timestamp == 0)
        {
            // 这种情况是需要过滤的
            return;
        }
        // 如果是第一条formatDescription忽略
        // 后面的formatDescription暂时没发现问题
        if (connectionAttributes.isFirstFormatDescription())
        {
            connectionAttributes.setFirstFormatDescription(false);
            return;
        }
        // 修复query|ROWS_QUERY中时间戳为0的情况
        if (EventType.query == eventHeader.getEventType() || EventType.rows_query == eventHeader.getEventType())
        {
            if (0 == timestamp)
            {
                return;// 直接忽略
            }
        }
        // 可以处理了
        try
        {
            String day = TimeUtils.getTimeDay(timestamp);
            log.info("snapShotType:" + eventHeader.getEventType());
            log.info("timestamp:" + timestamp + " day:" + day);
            connectionAttributes.setCurrentPosition(connectionAttributes.getNextPosition());
            connectionAttributes.setNextPosition(eventHeader.getNextPosition());
            // 第一次进来先记下当前天
            if (connectionAttributes.getHistoryPositionDay() == null)
            {
                connectionAttributes.setHistoryPositionDay(day);
            } else if (!day.equals(connectionAttributes.getHistoryPositionDay()))
            {
                // 需要记录到zk
                log.info("write history record: " + day + " --- " + connectionAttributes.getBinlogFileName()
                        + ":" + connectionAttributes.getCurrentPosition());
                String historyPath = connectionAttributes.getBinlogPositionPath() + "/" + day;
                if (!ZooKeepers.exist(historyPath))
                {
                    ZooKeepers.createPersistent(historyPath, connectionAttributes.getBinlogFileName() + ":"
                            + connectionAttributes.getCurrentPosition() + ":" + System.currentTimeMillis());
                } else
                {
                    ZooKeepers.update(historyPath, connectionAttributes.getBinlogFileName() + ":"
                            + connectionAttributes.getCurrentPosition() + ":" + System.currentTimeMillis());
                }
                log.info("write record finish...");
                connectionAttributes.setHistoryPositionDay(day);
                // 并判断zk是否保存了30天的数据，如果超过则删除
                List<String> children = ZooKeepers.getChildrenList(connectionAttributes.getBinlogPositionPath());
                for (String child : children)
                {
                    if (((timestamp - TimeUtils.getDayTime(child))
                            / Constants.millisecond_of_one_day) > Constants.history_record_day)
                    {
                        ZooKeepers.deletePath(connectionAttributes.getBinlogPositionPath() + "/" + child);
                    }
                }
            }
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        log.info("leave handleZKHistoryRecord");
        // LoggerUtils.debug(logger, "");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception
    {
        // 强调的一点是：到这个类之后，才会跟之前初始化的任务线程发生关系，否则有问题
        log.info("------------------------------------------");
        log.info("BinlogEventParseChannelInboundHandler - channelRead0 enter...");
        // LoggerUtils.info(logger, msg.toString());
        // 如何来做并行加速，就是把数据相关的东西提前取出来，抛到消费线程组里
        try
        {
            EventHeader eventHeader = new EventHeader();
            eventHeader.setTimestamp(ByteUtils.readUnsignedLong(byteBuf, 4) * 1000L);
            eventHeader.setEventType(EVENT_TYPES[ByteUtils.readUnsignedInt(byteBuf, 1)]);
            eventHeader.setServerId(ByteUtils.readUnsignedLong(byteBuf, 4));
            eventHeader.setEventLength(ByteUtils.readUnsignedLong(byteBuf, 4));
            eventHeader.setNextPosition(ByteUtils.readUnsignedLong(byteBuf, 4));
            eventHeader.setFlags(ByteUtils.readUnsignedInt(byteBuf, 2));
            log.info(eventHeader.toString());
            // 2)处理
            // 决定是否更新binlog[name:position]的历史记录
            ConnectionAttributes connectionAttributes = ((NioSocketChannelExtend) channelHandlerContext.channel()).getConnectionAttributes();
            // 一天才会发生一次，所以留在这里不会影响太大
            handleZKHistoryRecord(eventHeader, connectionAttributes);
            log.info("(((snapShotType: " + eventHeader.getEventType());
            // --------------------
            // 3)获取EventParser
            EventDataParser eventDataParser = eventTypeToEventDataParserMap.get(eventHeader.getEventType());
            if (null == eventDataParser)
            {
                // 找不到的事件就不要处理了
                // 但是最好打个日志
                log.info("parser not found..." + eventHeader.getEventType());
            } else
            {
                int checksumLength = (int) connectionAttributes.getChecksumLength();
                EventData eventData = eventDataParser.parse(byteBuf, channelHandlerContext, checksumLength);
                // LoggerUtils.debug(logger, eventData.toString());
                // 4)关联
                eventData.setEventHeader(eventHeader);
                // 5)处理各种事件
                handle(eventHeader.getEventType(), eventHeader, eventData, connectionAttributes, channelHandlerContext);
            }

        } catch (Exception e)
        {
            log.info(e.toString());
            throw new Exception(e);
        }
        log.info("BinlogEventParseChannelInboundHandler - channelRead0 finished...");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable)
    {
        // Close the connection when an exception is raised.
        // cause.printStackTrace();//务必要关闭
        log.info(throwable.toString());
        Nettys.cleanChannelContext(channelHandlerContext, throwable);
    }

    @Override
    public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
    {
        log.info("[channelInactive] socket is closed by remote server");
        Nettys.cleanChannelContext(channelHandlerContext, null);
    }

}
